package com.atguigu.userprofile.app

import com.atguigu.userprofile.common.bean.TagInfo
import com.atguigu.userprofile.common.dao.TagInfoDAO
import com.atguigu.userprofile.common.util.{ClickhouseUtil, MyPropertiesUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

import java.util.Properties


object TaskExportCkApp {

  //离线数仓 能不能保证幂等性 失败重试会不会出现重复？  overwrite
  // 离线批处理 幂等性是非常好 界定数据范围的
  // 实时计算是 失败重试 是非常不好界定数据范围
  // 画像 任务 失败重试  很好界定  重跑之前把数据删掉 就可以  不需要额外的去重技术来处理幂等问题

  // 任务三 ： 把hive中的宽表平移到clickhouse中
  //1  在clickhouse中动态建表
  //
  //2  读取hive的数据 ，读取成什么呢？ DataFrame
  //
  //3  写入clickhouse      可以通过df自带的jdbc写入方法完成
  def main(args: Array[String]): Unit = {

    val taskId = args(0)
    val busiDate = args(1)
    val sparkConf: SparkConf = new SparkConf().setAppName("task_export_ck_app")//.setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()

    val properties: Properties = MyPropertiesUtil.load("config.properties")
    val CLICKHOUSE_URL = properties.getProperty("clickhouse.url")
    val upDbName: String = properties.getProperty("user-profile.dbname")

    //1  在clickhouse中动态建表
    //   create table tablename_busidate
    //   (uid String ,tagcode1 String ,tagcode2 String ....)
    //  engine=ReplactedReplacingMerge  MergeTree  //高可用   去重 //精确一次  幂等**  //分布式？distribute
    // 基于画像的数据库 单点完全可以 //
    // partition by  不需要
    // primary key uid  和orderby 相同 就可以省略
    //  order by  uid
    //  8192
    val tagInfoList: List[TagInfo] = TagInfoDAO.getTagInfoWithOnList()
    val tagCodeSQL: String = tagInfoList.map(tagInfo => s"${tagInfo.tagCode.toLowerCase} String").mkString(",")

    val tableName = s"user_tag_merge_${busiDate.replace("-","")}"

    val dropTableSQL = s"drop table  if exists $tableName"

    val createTableSQL =
      s"""
         |    create table if not exists $tableName
         |      (uid String ,$tagCodeSQL )
         |      engine=MergeTree
         |      order by  uid
         |""".stripMargin


    println(dropTableSQL)
    ClickhouseUtil.executeSql(dropTableSQL)  //driver
    println(createTableSQL)
    ClickhouseUtil.executeSql(createTableSQL)//driver


    //2  读取hive的数据(宽表) ，读取成什么呢？ DataFrame
    val dataFrame: DataFrame = sparkSession.sql(s"select * from  $upDbName.$tableName")


    //3  写入clickhouse      可以通过df自带的jdbc写入方法完成
//    val rows: Array[Row] = dataFrame.rdd.collect()
    // 执行的位置 driver   会把ex中的数据提取到dr中 ，  1  io负担 2 driver内存爆炸 3 写入操作变为单线程
//    for (row <- rows ) {
//      // 一条条写入数据库
//      //  每1千条写入一次
//      ClickhouseUtil.executeSql(...)
//    }


    //执行的位置
    dataFrame.write.mode(SaveMode.Append)
      .option("batchsize",1000)
      .option("isolationLevel","NONE")
      .option("numPartitions","4")
      .option("driver","ru.yandex.clickhouse.ClickHouseDriver")
      .jdbc(CLICKHOUSE_URL,tableName,new Properties() )

  }

}
